package com.bossien.flink.common;

import com.alibaba.fastjson.JSONObject;
import com.bossien.common.model.student.ExamQuestionAnswer;
import com.bossien.common.utils.StringUtils;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.experimental.Accessors;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;

import java.io.Serializable;
import java.lang.reflect.Field;

/***
 * @author 胡春林
 * json串转pojo(一参数转一参数,二元数组对象)
 */
@Data
@Accessors(chain = true)
@Slf4j
public class PoJoTupleMapFunction<T> implements MapFunction<String,Tuple2<String,T>> {

    //需要转换的目标对象
    private Class<T> t;

    //用于Tuple2的f0值
    private String key;

    public PoJoTupleMapFunction(Class<T> t, String key){

        this.t = t;
        this.key = key;
    }


    @Override
    public Tuple2<String,T> map(String s) throws Exception {

        if (t == null)
            throw new Exception("没有可用于转换的对象，请确认你的参数");
        Tuple2<String, T> tuple2 = new Tuple2<>();
        try {
           T aClass = JSONObject.parseObject(s, t);
            //获取对应的变量名称
           Field field = aClass.getClass().getDeclaredField(getKey());
            if(null == field || StringUtils.isBlank(field.getName()))
            throw new Exception("请确认您的key，该对象没有此字段。");
            field.setAccessible(true);
            tuple2.f0 = (String) field.get(field.getName());
            tuple2.f1 = aClass;
        }catch (Exception e)
        {
            log.info("json串转序列化对象（二元数组）失败，请确认你的构造函数和相关参数");
        }
        return tuple2;
    }
}
